package com.glink.manage.mqtt;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

/**
 * Created by IntelliJ IDEA.
 *
 * @Author : qiushaoshan
 * @create 2024/7/30 16:06
 */
@Component
@Slf4j
public class MqttServiceImpl implements MqttService{

    @Resource
    private MqttProperties properties;
    
    @Resource
    private MqttSubscribeProperties mqttSubscribeProperties;

    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private MqttClient sendMqttClient;

    @PostConstruct
    public void init() {
        log.info("mqtt初始化{}", this.properties.getBroder());
        this.sendMqttClient = createMqttClient();
        (new Thread(this::subscribe)).start();
    }

    public void subscribe() {
        if (this.sendMqttClient != null && this.sendMqttClient.isConnected()) {
            try {
                List<String> productKeyList = this.mqttSubscribeProperties.getCarProductKeyList();
                if(CollectionUtils.isNotEmpty(productKeyList)){
                    for (String productKey : productKeyList) {
                        this.sendMqttClient.subscribe(productKey + "/+/" + this.mqttSubscribeProperties.getSubscribe(), this.properties.getQos());
                    }
                }

                this.sendMqttClient.subscribe(this.mqttSubscribeProperties.getSubscribe(), this.properties.getQos());
                this.sendMqttClient.subscribe(this.mqttSubscribeProperties.getCameraRegisterSubscribe(), this.properties.getQos());
            } catch (MqttException e) {
                log.error("mqtt订阅异常:{}, {}", this.mqttSubscribeProperties.getCarProductKeyList(), this.mqttSubscribeProperties.getSubscribe(), e);
            }
        }
    }
    
    private synchronized MqttClient createMqttClient() {
        MemoryPersistence persistence = new MemoryPersistence();
        MqttConnectOptions connOpts = new MqttConnectOptions();
        try {
            MqttClient mqttClient = new MqttClient(this.properties.getBroder(), this.properties.getClientId(), persistence);

            connOpts.setAutomaticReconnect(true);
            connOpts.setUserName(this.properties.getUserName());
            connOpts.setPassword(this.properties.getPassword().toCharArray());
            connOpts.setCleanSession(false);
            connOpts.setConnectionTimeout(this.properties.getTimeout());
            connOpts.setMaxInflight(this.properties.getMaxInflight());
            mqttClient.setTimeToWait(this.properties.getTimeToWait());

            mqttClient.setCallback(new AIMqttCallback());
            mqttClient.connect(connOpts);
            return mqttClient;
        } catch (MqttException e) {
            log.error("mqtt创建连接失败", e);

            return null;
        }
    }
    @Override
    public void sendMessage(String topic, Object obj) {
        if (topic == null || obj == null) {
            return;
        }
        String content = JSON.toJSONString(obj);
        this.threadPoolTaskExecutor.submit(() -> {
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(this.properties.getQos());
            try {
                if (!checkRecreateClient()) {
                    log.error("mqtt发送消息失败，重新创建连接失败");
                    return;
                }
                this.sendMqttClient.publish(topic, message);
            } catch (MqttException e) {
                log.error("mqtt发送消息失败{}", content);
                log.error("mqtt发送消息异常", e);
            }
        });
        log.info("发送mqtt消息,topic:{} content:{}", topic, content);
    }

    public boolean checkRecreateClient() {
        if (Objects.isNull(this.sendMqttClient) || !this.sendMqttClient.isConnected()) {
            log.info("mqttClient不存在或已断开，重新创建连接");
            this.sendMqttClient = createMqttClient();
            subscribe();
            return (Objects.nonNull(this.sendMqttClient) && this.sendMqttClient.isConnected());
        }
        return true;
    }
    
    @Override
    public void disconnection() {
        if (this.sendMqttClient != null && this.sendMqttClient.isConnected()) {
            try {
                this.sendMqttClient.disconnect();
            } catch (MqttException e) {
                log.error("mqtt断开连接时发生异常", e);
            }
        }
    }
}
